package cn.doitedu.eagle.functions;

import cn.doitedu.eagle.beans.*;
import cn.doitedu.eagle.names.MetricNames;
import cn.doitedu.eagle.queryservice.FlinkStateBufferManager;
import cn.doitedu.eagle.queryservice.UserActionCountQueryServiceImpl;
import cn.doitedu.eagle.queryservice.UserActionSequenceQueryService;
import cn.doitedu.eagle.queryservice.UserActionSequenceQueryServiceImpl;
import cn.doitedu.eagle.ruleutils.RuleHandler;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.kie.api.runtime.KieSession;

import java.util.List;
import java.util.Map;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/3/15
 * 规则计算核心代码
 **/
public class RuleProcessFunction extends KeyedBroadcastProcessFunction<String, LogBean, String, String> {

    // 规则事件阈值条件查询服务
    private UserActionCountQueryServiceImpl userActionQueryService;

    // 规则事件序列条件查询服务
    private UserActionSequenceQueryService userActionSequenceQueryService;

    // 事件流缓存state
    private ListState<LogBean> eventState;

    // drools规则kieSession对象缓存state
    private final MapStateDescriptor<String, RuleKieSessionStateBean> ruleStateDesc;

    // 存放 count条件查询结果的缓存buff：desc 和 state
    private final MapStateDescriptor<String, String> eventCountBuffStateDesc;
    private MapState<String, String> eventCountBuffState;


    // 存放 seq条件查询结果的缓存buff：desc 和 state
    private final MapStateDescriptor<String, String> ruleEventSeqBuffStateDesc;
    private MapState<String, String> ruleEventSeqBuffState;


    // 存放每个用户-每个规则-触发且命中次数
    private final MapStateDescriptor<String, Integer> trigHitCountStateDesc;
    private MapState<String, Integer> trigHitCountState;

    // 存放每个用户-每个规则-触发且命中次数 测流输出标记
    private final OutputTag<String> deviceRuleTrigHitCountTag;

    // 存放每个用户-每个规则-触发次数
    private MapState<String, Integer> trigCountState;

    // 侧流输出标记（输出所有事件，无论触发规则）
    private final OutputTag<LogBean> event_outs;

    // 存放每个用户-每个规则-触发次数 测流输出标记
    private final OutputTag<String> deviceRuleTrigCountTag;

    // metrics度量
    Counter onlineRuleSCounter;


    public RuleProcessFunction(MapStateDescriptor<String, RuleKieSessionStateBean> ruleStateDesc,
                               OutputTag<LogBean> event_outs,
                               MapStateDescriptor<String, Integer> trigHitCountStateDesc,
                               OutputTag<String> deviceRuleTrigHitCountTag,
                               OutputTag<String> deviceRuleTrigCountTag,
                               MapStateDescriptor<String, String> eventCountBuffStateDesc,
                               MapStateDescriptor<String, String> ruleEventSeqBuffStateDesc) {
        this.event_outs = event_outs;
        this.ruleStateDesc = ruleStateDesc;
        this.trigHitCountStateDesc = trigHitCountStateDesc;
        this.deviceRuleTrigHitCountTag = deviceRuleTrigHitCountTag;
        this.deviceRuleTrigCountTag = deviceRuleTrigCountTag;
        this.eventCountBuffStateDesc = eventCountBuffStateDesc;
        this.ruleEventSeqBuffStateDesc = ruleEventSeqBuffStateDesc;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        /*
         * 创建事件缓存state，并设置TTL为2小时
         * **/
        ListStateDescriptor<LogBean> eventsStateDesc = new ListStateDescriptor<>("events_state", LogBean.class);
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();
        eventsStateDesc.enableTimeToLive(ttlConfig);
        eventState = getRuntimeContext().getListState(eventsStateDesc);

        // 创建触发且命中计数state
        trigHitCountState = getRuntimeContext().getMapState(trigHitCountStateDesc);

        // 创建触发计数state
        MapStateDescriptor<String, Integer> trigCountStateDesc = new MapStateDescriptor<>("trigCountStateDesc", String.class, Integer.class);
        trigCountState = getRuntimeContext().getMapState(trigCountStateDesc);

        eventCountBuffState = getRuntimeContext().getMapState(eventCountBuffStateDesc);
        ruleEventSeqBuffState = getRuntimeContext().getMapState(ruleEventSeqBuffStateDesc);

        // 构建行为属性条件查询服务
        FlinkStateBufferManager flinkStateBufferManager = new FlinkStateBufferManager(eventCountBuffState, ruleEventSeqBuffState);
        userActionQueryService = new UserActionCountQueryServiceImpl(getRuntimeContext(),eventState, flinkStateBufferManager);

        // 构建行为序列条件查询服务
        userActionSequenceQueryService = new UserActionSequenceQueryServiceImpl(getRuntimeContext(),eventState, flinkStateBufferManager);

        // metrics度量
        onlineRuleSCounter = getRuntimeContext().getMetricGroup().counter(MetricNames.ONLINE_RULE_CNTS);

    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void processElement(LogBean value, ReadOnlyContext ctx, Collector<String> out) throws Exception {

        // 将本条事件，存入state
        eventState.add(value);

        /*
         * 从广播状态中，获取规则
         * **/
        Iterable<Map.Entry<String, RuleKieSessionStateBean>> ruleStateBeans = ctx.getBroadcastState(ruleStateDesc).immutableEntries();
        // 如果kieSessions列表为空，则说明尚没有规则，啥也不做
        for (Map.Entry<String, RuleKieSessionStateBean> ruleStateBeanEntry : ruleStateBeans) {
            /*
             * 获取本条规则的ruleId
             * **/
            String ruleId = ruleStateBeanEntry.getKey();
            /*
             * 获取drools规则运算对象
             * **/
            RuleKieSessionStateBean stateBean = ruleStateBeanEntry.getValue();
            KieSession kieSession = stateBean.getKieSession();

            /*
             * 构造一个"触发条件事件参数"对象
             * 下面几个对象，是逐层包装
             * **/
            RuleParamEvent triggerEvent = new RuleParamEvent();
            triggerEvent.setEventId(value.getEventId());
            triggerEvent.setProperties(value.getProperties());

            /*
             * 构造一个“规则整体参数”对象
             * **/
            RuleParam ruleParam = new RuleParam();
            ruleParam.setEventCountQuerySqls(stateBean.getEventCountQuerySql()); // 设置规则对应的count查询sql
            ruleParam.setEventSeqQuerySql(stateBean.getEventSeqQuerySql()); // 设置规则对应的序列查询sql

            ruleParam.setRuleId(ruleId);  // 向规则参数中设置: 规则ruleId
            ruleParam.setDeviceId(value.getDeviceId());  // 向规则参数中设置: 当前用户id
            ruleParam.setTrigEvent(triggerEvent);   // 向规则参数中设置: 触发条件事件

            /*  从触发次数统计state中获取当前用户当前规则的已命中次数，用于规则中判断是否已经到达命中次数上限  **/
            Integer trigDone = trigHitCountState.get(ruleStateBeanEntry.getKey());
            ruleParam.setTrigDone(trigDone == null ? 0 : trigDone);

            /*
             *  构造一个 "drools交互对象"
             *  用于传给drools后，注入"规则整体参数"，及注入规则判断结果
             * **/
            DroolRuleParam droolRuleParam = new DroolRuleParam();
            droolRuleParam.setRuleParam(ruleParam);  // 向交互对象中：加入“规则整体参数”对象
            droolRuleParam.setTrigEventId(value.getEventId());  // 向交互对象中：加入触发条件事件的eventId

            droolRuleParam.setUserActionCountQueryService(userActionQueryService); // 向交互对象中：加入查询service对象
            droolRuleParam.setUserActionSequenceQueryService(userActionSequenceQueryService);


            /*
             * 核心代码：启动规则运算
             * 如果已经达到规则限制的最大触发次数条件，则不再进行规则匹配
             * **/
            kieSession.insert(droolRuleParam);  // 插入"drools交互对象"
            int trigCnt = kieSession.fireAllRules(); // 触发规则运算的条数

            // 更新触发次数计数器
            trigCountState.put(ruleId, trigCountState.get(ruleId) == null ? 0 : trigCountState.get(ruleId) + trigCnt);

            ctx.output(deviceRuleTrigCountTag, droolRuleParam.getRuleParam().getDeviceId() + "," + ruleId + "," + trigCountState.get(ruleId));

            /*
             * 从"drools交互对象"中获取规则运算结果，如果命中，则输出推送结果
             * **/
            if (droolRuleParam.getRuleParam().isHit()) {
                /*
                 * 更新触发次数统计计数器
                 * **/
                int hitCnt = ruleParam.getTrigDone() + 1;
                trigHitCountState.put(ruleId, hitCnt);

                /*
                 * 输出规则触发推送结果
                 * **/
                out.collect(new RuleResult(droolRuleParam.getRuleParam().getRuleId(), droolRuleParam.getRuleParam().isHit(), droolRuleParam.getTrigEventId(), droolRuleParam.getRuleParam().getDeviceId()) + value.getProperties().toString());

                /*
                 * 侧流输出规则触发统计key级别局部结果
                 * **/
                ctx.output(deviceRuleTrigHitCountTag, droolRuleParam.getRuleParam().getDeviceId() + "," + ruleId + "," + hitCnt);
            }
        }
        // 侧流输出全部事件（用于全量落盘）
        ctx.output(event_outs, value);
    }

    @Override
    public void processBroadcastElement(String canalStr, Context ctx, Collector<String> out) {

        // 解析 binlog
        CanalBinlog canalBinlog = JSON.parseObject(canalStr, CanalBinlog.class);
        if (StringUtils.isNotBlank(canalBinlog.getType()) &&
                (canalBinlog.getType().equals("INSERT") || canalBinlog.getType().equals("UPDATE") || canalBinlog.getType().equals("DELETE"))) {
            // 获取数据内容
            List<CanalRuleRecord> data = canalBinlog.getData();
            CanalRuleRecord canalRuleRecord = data.get(0);

            // 获取ruleId
            String ruleId = canalRuleRecord.getRule_name();
            // 通过rule状态来判断控制类型： 1：新增或修改  2：删除  3：停用  4：激活
            String command = canalRuleRecord.getRule_status();
            // 获取rule代码
            String ruleCode = canalRuleRecord.getRule_code();
            // 获取rule 事件阈值 查询sql
            String eventCountQuerySql = canalRuleRecord.getEventCountQuerySqls();
            // 获取rule 事件序列  查询sql
            String eventSeqQuerySql = canalRuleRecord.getEventSeqQuerySql();

            // 调用工具，处理进来的规则代码（新增，修改，删除，启用，停用）
            RuleHandler.handleRule(
                    ctx.getBroadcastState(ruleStateDesc),
                    ruleId,
                    command,
                    ruleCode,
                    eventCountQuerySql,
                    eventSeqQuerySql,onlineRuleSCounter);
        }
    }

}
